package com.wu.gconduit.realtime.app

import com.alibaba.fastjson.JSON
import com.wu.gconduit.common.constant.GconduitConstant
import com.wu.gconduit.common.util.ESUtil
import com.wu.gconduit.realtime.bean.{ConduitAlertInfo, EventLog}
import com.wu.gconduit.realtime.util.MyKafkaUtil
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}

import java.util
import scala.util.control.Breaks._


/**
 * @author Wumeng
 * @Date 2022年01月11日 17:58
 *       预警
 */
object AlterAppConduit {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("AlterApp")
    val ssc = new StreamingContext(conf, Seconds(3))
    //     1、从kafka中消费数据(需要运行kafka,日志服务器，nginx，实时的JsonMock)
    val sourceStream = MyKafkaUtil
      .getKafkaStream(ssc, GconduitConstant.TOPIC_EVENT) //获取Kafka中的数据

      .window(Minutes(5), Seconds(6)) //添加窗口，设置窗口的大小
    //    2、数据封装
    val eventLogStream: DStream[EventLog] = sourceStream.map(s => JSON.parseObject(s, classOf[EventLog]))

    //    eventLogStream.print(1000)

    //    3、实现需求
    //    3.1、同一设备（按照mid进行分组）
    val eventLogGrouped: DStream[(String, Iterable[EventLog])] = eventLogStream
      .map(eventlog => (eventlog.mid, eventlog))
      .groupByKey()

    //    eventLogGrouped.print(1000)
    //3.2、产生预警信息
    val alertInfoStream = eventLogGrouped.map {
      //3、三次以上使用不同的用户登录
      //4、领取优惠卷
      //5、在领卷过程中，没有浏览商品
      case (mid, eventlogs: Iterable[EventLog]) =>
        //注意，需要使用java集合，把集合中的数据写入到es时，scala集合不支持（不能取出集合数据）
        //a存储登录过的所有用户，统计当前的设备id，最近5分钟登录过的所有用户
        val uidSet: util.HashSet[String] = new util.HashSet[String]()

        //b存储5分钟内当前设备所有的事件
        val eventList: util.ArrayList[String] = new util.ArrayList[String]()

        //c存储领取的优惠卷对应的产品id
        val itemSet: util.HashSet[String] = new util.HashSet[String]()

        //d是否点击商品，默认false
        var isClickItem: Boolean = false
        breakable {
          eventlogs.foreach(log => {
            eventList.add(log.eventId) //存储所有事件
            log.eventId match {
              case "coupon" =>
                uidSet.add(log.uid) //存储领取优惠卷的用户id
                itemSet.add(log.itemId) //存储优惠卷，对应的商品id
              case "clickItem" =>
                //只要有一次浏览商品，就不因该产生预警信息
                isClickItem = true
                //这里不要使用return，使用return是退出最外层函数
                break
              case _ => //其他事件忽略
            }
          })

        }
        //（boolean，预警信息）
        //（是否需要预警信息，预警信息）
        //设备数大于3，并且没有浏览过商品
        /*
        mid, uidSet：当前的设备id, itemSet：取的优惠卷对应的产品id, eventList:事件, System.currentTimeMillis()：时间戳
        (是否需要预警，预警信息)
         */
        //        (uidSet.size() > 3 && !isClickItem, AlertInfo(mid, uidSet, itemSet, eventList, System.currentTimeMillis()))
        (uidSet.size() >= 2 && !isClickItem, ConduitAlertInfo(mid,  System.currentTimeMillis()))
    }

    alertInfoStream.
      filter(_._1).
      map(_._2).
      foreachRDD(rdd => {
        rdd.foreachPartition((alterInfos: Iterator[ConduitAlertInfo]) => {
          //连接es
          //写数据
          //关闭连接
          //          同一设备每分钟只记录一次预警。spark-streaming下不实现，交给es自己来实现；在es中，id每分钟变化一次
          //          针对每个设备，一分钟只能产生一条信息，以分钟作为Id,每分钟才会变一次，一分钟如果写多次，只会保留最后一个
          //          每个设备都可能会报警，所以需要把mid体现出来
          val data: Iterator[(String, ConduitAlertInfo)] = alterInfos
            .map(info => (info.mid + ": " + info.ts / 1000 / 60, info))

          //        ESUtil.insertBulk("gmall_coupon_my_alert",data)
          ESUtil.insertBulk(GconduitConstant.INDEX_MY_ALTER_COPY, data)

        })
      })

    alertInfoStream.print(10000)

    ssc.start()
    ssc.awaitTermination()
  }



  /*
  1.	同一设备(按照设备分组)
  2.	5分钟内(window)
  3.	三个不同账号登录(统计登录的用户，聚合)
  4.	领取优惠券（过滤）
  5.	并且没有浏览商品
  6.	同一设备每分钟只预警一次(同一设备, 每分钟只向 es 写一次记录，在es中，id每分钟变化一次)

   */
}
